package momo.app.comms;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import momo.app.util.FileOutput;
import momo.app.util.IntByteConvert;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class SendChannel extends Thread
{
	private Log log = LogFactory.getLog(SendChannel.class);
	private Log logStructure = LogFactory.getLog("MoMoStructureLog");
	
	private final static int timeToWait = 30;
	
	private InetAddress ip;
	private int commPort;
	private LinkedBlockingQueue<String> queue;
	
	private Socket sock;
	private BufferedOutputStream out;
	
	private String hisIP;
	private int myPort;
	
	private boolean interrupted;
	
	public SendChannel(InetAddress ip, int commPort) throws IOException
	{
		this.ip = ip;
		this.commPort = commPort;
		queue = new LinkedBlockingQueue<String>();
		
		sock = new Socket();
		sock.connect(new InetSocketAddress(ip, commPort));
		hisIP = sock.getInetAddress().getHostAddress();
		myPort = sock.getLocalPort();
		out = new BufferedOutputStream(sock.getOutputStream());
		startReadThread();
		this.start();
	}
	
	public void run()
	{
		if ( log.isDebugEnabled() )
			log.debug("SendChannel ("+myPort+") STARTS for "+hisIP+":"+commPort);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("SendChannel ("+myPort+") STARTS for "+hisIP+":"+commPort);
		
		while(!interrupted)
		{
			try
			{
				String data = queue.poll(timeToWait, TimeUnit.SECONDS);
				
				if ( data != null )
					directSending(data);
				
			}catch (InterruptedException ie)
			{
				if ( log.isErrorEnabled() )
					log.error("InterruptedException for SendChannel ("+myPort+") of "+hisIP+":"+commPort, ie);
				stopSending();
			}catch(Exception e)
			{
				if ( log.isErrorEnabled() )
					log.error("Exception for SendChannel ("+myPort+") of "+hisIP+":"+commPort, e);
				stopSending();
			}
		}
		
		try
		{
			out.close();
			sock.close();
		}catch (IOException ioe)
		{
			if ( log.isErrorEnabled() )
				log.error("IOException", ioe);
		}
		
		if ( log.isDebugEnabled() )
			log.debug("SendChannel ("+myPort+") ENDS for "+hisIP+":"+commPort);
		if ( logStructure.isDebugEnabled() )
			logStructure.debug("SendChannel ("+myPort+") ENDS for "+hisIP+":"+commPort);
	}
	
	private void directSending(String data) throws IOException
	{
		if ( log.isDebugEnabled() )
			log.debug("Sending data ("+myPort+") to " +ip+":"+commPort+"\n"+data);
		
		byte[] byteData = IntByteConvert.dataToByteArray(data); 
		out.write(byteData);
		out.flush();
	}
	
	public void startReadThread()
	{
		Thread readThread = new Thread()
		{
			public void run()
			{
				try {
//					BufferedReader in = new BufferedReader(new InputStreamReader(sock.getInputStream()));
//					
//					in.readLine();
					
					BufferedInputStream in = new BufferedInputStream(sock.getInputStream());
					in.read();
					
				} catch (IOException e)
				{
					//will not be catching anything
				}
				
				if ( log.isDebugEnabled() )
					log.debug("Sending Thread ("+myPort+") ends by rcving side");
				if ( logStructure.isDebugEnabled() )
					logStructure.debug("Sending Thread ("+myPort+") ends by rcving side");
				
				stopSending();
			}
		};
		readThread.start();
	}
	
	public void stopSending()
	{
		interrupted = true;
	}
	
	public void shutdownNow()
	{
		stopSending();
		try
		{
			sock.close();
		}catch(IOException ioe)
		{
			if ( log.isErrorEnabled() )
				log.error("IOException", ioe);
		}
	}
	
	public int getMyPort()
	{
		return myPort;
	}
	
	public void put(String data)
	{
		try
		{
			queue.put(data);
			if ( log.isDebugEnabled() )
				log.debug("("+myPort+")Data: "+data+", put in Send Buffer of commPort: " + commPort);
		}catch(InterruptedException ie)
		{
			if ( log.isErrorEnabled() )
				log.error("InterruptedException", ie);
		}
	}
	
	public String toString()
	{
		StringBuffer buffer = new StringBuffer();
		buffer.append("SendChannel: ");
		buffer.append(ip);
		buffer.append(":");
		buffer.append(commPort);
		buffer.append(":");
		buffer.append(myPort);
		return buffer.toString();
	}

	@Override
	public int hashCode() {
		final int prime = 31;
		int result = 1;
		result = prime * result + commPort;
		result = prime * result + ((hisIP == null) ? 0 : hisIP.hashCode());
		result = prime * result + (interrupted ? 1231 : 1237);
		result = prime * result + ((ip == null) ? 0 : ip.hashCode());
		result = prime * result + ((log == null) ? 0 : log.hashCode());
		result = prime * result + myPort;
		result = prime * result + ((out == null) ? 0 : out.hashCode());
		result = prime * result + ((queue == null) ? 0 : queue.hashCode());
		result = prime * result + ((sock == null) ? 0 : sock.hashCode());
		return result;
	}

	
}//end of class SendCommsChannel
